package org.example.source;

import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.client.avro.SimpleTextLineEventReader;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

import java.util.HashMap;
import java.util.SimpleTimeZone;

/**
 * 自定义Source
 * @ClassName MySource
 * @Author 小坏
 * @Date 2021/10/26、16:42
 * @Version 1.0
 */
public class MySource extends AbstractSource implements Configurable, PollableSource {


    private String prefix;
    private String subfix;
    private Long delay;

    @Override
    public void configure(Context context) {
        //前缀默认有一个
        prefix = context.getString("pre", "pre-");
        //后缀不指定就没了
        subfix = context.getString("sub");

        delay = context.getLong("delay",2000L);
    }

    @Override
    public Status process() throws EventDeliveryException {


      try {
          for (int i = 0; i < 5; i++) {
              SimpleEvent event = new SimpleEvent();

              final HashMap<String, String> header = new HashMap<>();
              event.setHeaders(header);
              event.setBody((prefix + "atguigu:" + i + subfix).getBytes());
              getChannelProcessor().processEvent(event);
          }
          Thread.sleep(delay);
          return Status.READY;
      }catch (Exception e){
          e.printStackTrace();
          return Status.BACKOFF;
      }
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }


}
